package cn.itcast.sink;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

/**
 * @author KTL
 * @version V1.0
 * @Package cn.itcast.sink
 * @date 2021/2/24 0024 17:13
 * @Copyright © 2015-04-29  One for each, and two for each
 *                  演示DataSream   自定义sink
 */
public class SinkDemo02 {
    public static void main(String[] args) throws Exception {
        //TODO 0.env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
        //TODO 1.source
        final DataStreamSource<Student> StudentDs= env.fromElements(new Student(null, "tonyma", 18));
        //TODO 2.transfromation
        //TODO 3.sink
        StudentDs.addSink(new MySinkFunction());

        //TODO 4.excute
        env.execute();
    }
    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class Student{
        private String  id ;
        private String name ;
        private int age ;
    }

    /**
     * 将数据sink到mysql中
     */
    public static class MySinkFunction extends RichSinkFunction<Student> {
        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            /**
             * 写mysql数据连接内容
             */
        }

        @Override
        public void invoke(Student value, Context context)throws  Exception {

        }

        @Override
        public void close() throws Exception {
            super.close();
        }
    }

}
